package com.atguigu.pro1

import java.sql.Timestamp

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * @description: xxx
 * @time: 2021/4/1 15:48
 * @author: baojinlong
 * */
object AdClickAnalysis {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 方便测试全局并行度为1
    env.setParallelism(1)
    // 设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 从文件中读取数据
    // 从文件中读取数据
    val inputStream: DataStream[String] = env.readTextFile("C:/codes/scala/FlinkTutorial/src/main/resources/AdClickLog.csv")
    // 转换成样例类,并提取时间戳和watermark
    val adLogStream: DataStream[AdClickLog] = inputStream
      .map(data => {
        val arr: Array[String] = data.split(",")
        AdClickLog(arr(0).toLong, arr(1).toLong, arr(2), arr(3), arr(4).toLong)
      })
      // 数据是升序排列的,默认watermark延时1ms
      .assignAscendingTimestamps(_.timestamp * 1000)


    // 插入异步过滤操作,并将有刷单行为的用户输出到侧输出流(黑名单报警)
    val filterBlackUserStream: DataStream[AdClickLog] = adLogStream
      // 当前用户当前广告进行分组
      .keyBy(data => (data.userId, data.adId))
      .process(new FilterBlackListUserResult(100))

    // 开窗聚合统计
    val adCountResultStream: DataStream[AdClickCountByProvince] = adLogStream
      .keyBy(_.province)
      .timeWindow(Time.minutes(5), Time.seconds(5))
      // 后面是全窗口函数的回调类,可以拿到window信息.
      .aggregate(new AdCountAgg, new AdCountWindowResult)

    adCountResultStream.print("adCountResultStream")
    // 显示侧输出流
    val blackListUserWarning: DataStream[BlackListUserWarning] = filterBlackUserStream.getSideOutput(new OutputTag[BlackListUserWarning]("warning"))
    blackListUserWarning.print("blackListUserWarning")

    env.execute("add count statistic")

  }

}

// 定义输入输出样例类
case class AdClickLog(userId: Long, adId: Long, province: String, city: String, timestamp: Long)

case class AdClickCountByProvince(windowEnd: String, province: String, count: Long)

class AdCountAgg extends AggregateFunction[AdClickLog, Long, Long] {
  override def createAccumulator(): Long = 0

  override def add(in: AdClickLog, acc: Long): Long = acc + 1

  override def getResult(acc: Long): Long = acc

  override def merge(acc: Long, acc1: Long): Long = acc + acc1
}

// 这是windowFunction就是apply如果是processWindowFunction则是process
class AdCountWindowResult extends WindowFunction[Long, AdClickCountByProvince, String, TimeWindow] {
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[AdClickCountByProvince]): Unit = {
    val windowEnd: String = new Timestamp(window.getEnd).toString
    val province: String = key
    val count: Long = input.head
    out.collect(AdClickCountByProvince(windowEnd, province, count))
  }
}

// 侧输出流黑名单报警信息样例类
case class BlackListUserWarning(userId: Long, adId: Long, msg: String)

// 自定义KeyedProcessFunction K, I, O
class FilterBlackListUserResult(maxCount: Int) extends KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog] {
  // 定义状态,保存用户对广告的点击量 lazy或者放在open生命周期里
  lazy val countState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("countState", classOf[Long]))
  // 每天0点定时清空状态的时间戳
  lazy val resetTimerTsState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("resetTimerTsState", classOf[Long]))
  // 标记当前用户是否已经进入黑名单
  lazy val isBlackState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("isBlackState", classOf[Boolean]))

  // 每来一条数据都进行一次处理
  override def processElement(i: AdClickLog, context: KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog]#Context, collector: Collector[AdClickLog]): Unit = {
    val curCount: Long = countState.value
    // 判断只要是第一个数据来了,直接注册0点的清空状态定时器
    if (curCount == 0) {
      // 拿到处理时间,天数*时间毫秒
      val ts: Long = (context.timerService.currentProcessingTime / (1000 * 60 * 60 * 24) + 1) * 24 * 60 * 60 * 1000 - 8 * 60 * 60 * 1000
      // 更新清空定时器时间
      resetTimerTsState.update(ts)
      // 注册定时器
      context.timerService.registerEventTimeTimer(ts)
    }
    if (curCount >= maxCount) {
      // 判断是否已经在黑名单里面,如果没有的话才输出到侧输出流
      if (!isBlackState.value) {
        isBlackState.update(true)
        context.output(new OutputTag[BlackListUserWarning]("warning"), BlackListUserWarning(i.userId, i.adId, "Click ad over" + maxCount + " times today."))
      }
      // 超过的数据就不要输出到流下游数据流中
      return
    }

    // 正常情况,count+1,然后将数据原样输出
    countState.update(curCount + 1)
    collector.collect(i)
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog]#OnTimerContext, out: Collector[AdClickLog]): Unit = {
    // 如果当前定时器触发的时间和需要清空的定时器是一个值
    if (timestamp == resetTimerTsState.value) {
      isBlackState.clear()
      countState.clear()
    }
  }
}